package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo12Plant {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("shebao")
      .config("spark.sql.shuffle.partitions", 1) //指定spark sql shuffle之后rdd的分区数
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._
    val userLowCarbon: DataFrame = spark
      .read
      .format("csv")
      .option("sep", "\t")
      .schema("user_id STRING, date_dt STRING ,low_carbon DOUBLE")
      .load("data/ant_user_low_carbon.txt")

    userLowCarbon.cache()

    val plantCarbon: DataFrame = spark
      .read
      .format("csv")
      .option("sep", "\t")
      .schema("plant_id STRING , plant_name STRING ,plant_carbon  DOUBLE")
      .load("data/ant_plant_carbon.txt")

    //取出胡杨所需要的能量
    val huYangPlantCarbon: Double = plantCarbon
      .where($"plant_name" === "胡杨")
      .head()
      .getAs[Double]("plant_carbon")

    //取出沙柳所需要的能量
    val shaLiuPlantCarbon: Double = plantCarbon
      .where($"plant_name" === "沙柳")
      .head()
      .getAs[Double]("plant_carbon")


    /**
     * 假设2017年1月1日开始记录低碳数据（user_low_carbon），假设2017年10月1日之前满足申领条件的用户都申领了一颗p004-胡杨，
     * 剩余的能量全部用来领取“p002-沙柳” 。 统计在10月1日累计申领“p002-沙柳” 排名前10的用户信息；
     * 以及他比后一名多领了几颗沙柳。 得到的统计结果如下表样式：
     *
     * unix_timestamp: 将时间字符串转换成时间戳
     * from_unixtime： 将时间戳转换成时间字符串
     */
    //取出2017年1月1号到10月1号所有的数据
    userLowCarbon
      //转换日期格式
      .withColumn("com_date", from_unixtime(unix_timestamp($"date_dt", "yyyy/M/d"), "yyyy-MM-dd"))
      .where($"com_date" >= "2017-01-01" and $"com_date" <= "2017-10-01")
      //计算用户总的能量
      .groupBy($"user_id")
      .agg(sum($"low_carbon") as "sum_low_carbon")
      //满足条件的用户都领了一个胡杨，剩余的能量
      .withColumn("sum_low_carbon_diff", $"sum_low_carbon" - huYangPlantCarbon)
      //计算剩余的能量可以领取的沙柳的数量
      .withColumn("num", floor($"sum_low_carbon_diff" / shaLiuPlantCarbon))
      //排序取前11
      .orderBy($"num".desc)
      .limit(11)
      //取后易名领取沙柳的数量
      .withColumn("next_num", lead($"num", 1, 0) over Window.orderBy($"num".desc))
      .withColumn("diff_num", $"num" - $"next_num")
      .limit(10)
      //关联用户信息别
      .join(userLowCarbon, "user_id")
      .select($"user_id", $"date_dt", $"low_carbon")
      .show()


    /**
     * 查询user_low_carbon表中每日流水记录，条件为： 用户在2017年，连续三天（或以上）的天数里，
     * 每天减少碳排放（low_carbon）都超过100g的用户低碳流水。 需要查询返回满足以上条件的user_low_carbon表中的记录流水。
     * 例如用户u_002符合条件的记录如下，因为2017/1/2~2017/1/5连续四天的碳排放量之和都大于等于100g：
     */
    userLowCarbon
      //转换时间格式
      .withColumn("com_date", from_unixtime(unix_timestamp($"date_dt", "yyyy/M/d"), "yyyy-MM-dd"))
      //取2017年的数据
      .where(substring($"date_dt", 1, 4) === "2017")
      //计算用户每天总的能量
      .groupBy($"user_id", $"com_date", $"date_dt")
      .agg(sum($"low_carbon") as "sum_low_carbon")
      //取出每天超过100的数据
      .where($"sum_low_carbon" > 100)
      //计算连续
      .withColumn("r", row_number() over Window.partitionBy($"user_id").orderBy($"com_date"))
      .withColumn("clazz", expr("date_sub(com_date,r)"))
      //计算连续的天数
      .withColumn("lianxu_num", count($"clazz") over Window.partitionBy($"user_id", $"clazz"))
      //取出3天或者3天以上的数据
      .where($"lianxu_num" >= 3)
      //关联获取流水数据
      .join(userLowCarbon, List("user_id", "date_dt"))
      //取出字段
      .select($"user_id", $"date_dt", $"low_carbon")
      .show(1000)

    while (true){}

  }

}
